package com.saic.data.main;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import javax.mail.Address;
import javax.mail.Authenticator;
import javax.mail.Message;
import javax.mail.Message.RecipientType;
import javax.mail.MessagingException;
import javax.mail.PasswordAuthentication;
import javax.mail.SendFailedException;
import javax.mail.Session;
import javax.mail.Transport;
import javax.mail.internet.AddressException;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;

import com.saic.data.table.TableAction;
import com.saic.data.table.TableComparator;
import com.saic.data.table.TableFile;
import com.saic.data.table.TableUtil;
import com.saic.data.util.DBUtil;
import com.saic.data.util.HbaseUtil;
import com.saic.data.util.SimpleMail;
import com.saic.data.util.ZookeeperUtil;

/**
 * @检查唯一性：每天运行一次:每天9：00运行一次，工作日运行
 * @author wangk
 */
public class CheckSum {

	public static void main(String[] args) {
		//获取查询日期范围
		Calendar calendar=Calendar.getInstance();   
		calendar.setTime(new Date());
		calendar.set(Calendar.MINUTE,30);
		calendar.set(Calendar.SECOND,30);
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
		String dateto = sdf.format(calendar.getTime());
		calendar.set(Calendar.DAY_OF_MONTH,calendar.get(Calendar.DAY_OF_MONTH)-1);
		String datefrom = sdf.format(calendar.getTime());
		//加载ZK配置信息
		try {
			ZookeeperUtil zooUtil;
			zooUtil = new ZookeeperUtil();
			Map<String, String> zMap = zooUtil.getProperties(zooUtil.rootpath);
			zooUtil.close();
			
			TableUtil.pm("CheckSum-main", "get Check list" );
			List<TableFile> list = getCheckList(datefrom,dateto,zMap);
			//List<TableFile> list = getCheckList("20161003060000","20161005060000",zMap);
			StringBuffer content = new StringBuffer() ;
			if (list.size()<1){
				content.append("期间内没有数据导入！");
			}else content = getReport(list);
			
			
			System.out.println(content.toString());
			List<String> cList = new ArrayList<String>();
			cList.add("wangk@paraview.cn");
			cList.add("wanghua@saicservice.com");
			cList.add("zhutingting@saicmotor.com");
			cList.add("zhangninglu@saicservice.com");
			SimpleMail.sendMail("tansheng01@saicmotor.com", cList,"Finance File Import Result-"+dateto.substring(0, 8) ,content.toString());
			
			} catch (Exception e) {
			// TODO Auto-generated catch block
			TableUtil.pm("CheckSum-main", "Error when CheckSum:"+e.getMessage());
			e.printStackTrace();
		}
			
		
	}
	public static StringBuffer getReport(List<TableFile> list){
		StringBuffer content = new StringBuffer();
		TableUtil.pm("CheckSum-getCheckList", list.size()+" file have been found, chect file");
		for(int i= list.size()-1 ;i>=0;i--){
			TableFile tFile = list.get(i);
			String mString ="";
			TableUtil.pm("CheckSum-getCheckList", "check file "+tFile.getFileName()+" , fileStats is "+tFile.gettAction().toString());
			if (tFile.gettAction()==TableAction.Done){
				TableUtil.pm("CheckSum-getCheckList", "file "+tFile.getFileName()+" check imported line");
				try {
					mString = "文件 "+tFile.gettConf().getFtpDir()+"/"+tFile.getFileName() +"传输完成。 数据条目检查:"+doCheck(list.get(i)) + "\r\n";	
				} catch (Exception e) {
					// TODO: handle exception
					TableUtil.pm("CheckSum-main", "Error when Check File"+e.getMessage());
					continue;
				}
			}
			if(mString.length()>0) {
				content.append(mString);
				content.append("\n");
			}
		}
		if (content.length()<1) content.append("没有发现导入成功的数据文件！");
		return content;
	}
	
	public static List<TableFile> getCheckList(String datefrom,String dateto,Map<String,String> zMap) throws IOException{
		TableUtil.pm("CheckSum-getCheckList", "get list from "+datefrom +" to "+dateto );
		List<TableFile> list = new ArrayList<TableFile>();
		//查询Hbase获取file列表
		ResultScanner rScanner = HbaseUtil.QueryByConditionValue(TableUtil.configtab, TableUtil.configtabFamily, "modifiedDate", datefrom, dateto);
		for(Result result : rScanner){
			if (result.isEmpty()) break;
			String fileStats = HbaseUtil.getFromResult(result, TableUtil.configtabFamily, "fileStats");
			if (fileStats.equals("Done")) {
			TableFile tFile = new TableFile(result, zMap);
			list.add(tFile);}
		}
		rScanner.close();
		TableComparator tc = new TableComparator();
		Collections.sort(list,tc); 
		return list;
	}

	public static String doCheck(TableFile table) throws IOException {
		String fileKey = table.getRowkey();
		TableUtil.pm("CheckSum-getCheckList", "find result by Filekey= "+fileKey );
		
		ResultScanner rs = HbaseUtil.QueryByCondition(table.gettConf().getHistoryHtbl(), fileKey+"0", fileKey+"9");
		DBUtil sqlUtil= new DBUtil();
		String resultS="";
		int successLineHbase = 0; 
		int successLineOracle = 0;
		int totalLine = 0;
		List<Put> puts=new ArrayList<Put>();
		for (Result r : rs) {
			totalLine++;
			String _fileKey = HbaseUtil.getFromResult(r, table.gettConf().dataFamily,"fileKey");
			String _fileName = HbaseUtil.getFromResult(r, table.gettConf().dataFamily,"fileName");
            if(_fileKey.compareTo(fileKey)==0 && _fileName.compareTo(table.getFileName())==0){
            	successLineHbase++;
            	Map<String, String> map = new HashMap<String, String>();
            	map.put(TableUtil.flag_check, "1");
            	Put put = HbaseUtil.getPutFromMap(table.gettConf().getHistoryHtbl(),
					table.gettConf().dataFamily,new String(r.getRow()), map);
            	puts.add(put);}
		}
		TableUtil.pm("CheckSum-docheck", "search line:"+totalLine);
		HbaseUtil.insertData(table.gettConf().getHistoryHtbl(), puts);
        
		successLineOracle = sqlUtil.queryCount("select count(*) from "+ table.gettConf().getOracleTbl()
				+" where custom12=\'" + fileKey + "\'"); 	
		if (successLineOracle == successLineHbase && successLineOracle > 0) {
			sqlUtil.update("update " + table.gettConf().getOracleTbl()
					+ " set custom10=1 where custom12=\'" + fileKey + "\'");
			TableUtil.pm("CheckSum-docheck","File: "+ table.getFileName()+" Success lineOracle: " + successLineOracle
				+ " Success lineHbase: " + successLineHbase);
			resultS = "数据条目匹配，检测结果Hbase记录  "+successLineHbase+" 条，Oracle写入  "+successLineOracle+" 条。";
		}
		else {
			if(successLineOracle < 1){			
				TableUtil.pm("CheckSum-docheck", "Error: CheckSum-->Not in ORACLE");
				resultS = "Oracle 写入失败！下次运行将重新传输文件。";}
			else{
				TableUtil.pm("CheckSum-docheck","File: "+ table.getFileName()+" Success lineOracle: " + successLineOracle
					+ " Success lineHbase: " + successLineHbase);
				resultS = "数据条目不匹配，检测结果Hbase记录  "+successLineHbase+" 条，Oracle写入  "+successLineOracle+" 条。下次运行将重新传输文件。";
			}
			HbaseUtil.updateData(TableUtil.configtab, fileKey.getBytes(), 
				TableUtil.configtabFamily, "fileStats", TableAction.UploadRecord.toString());}
		return resultS;
	}
}
